package com.gzyj.flink.gps.map;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.serializer.SerializerFeature;
import com.alibaba.fastjson.support.spring.FastJsonRedisSerializer;
import com.gzyj.flink.gps.GPSPosition;
import com.gzyj.flink.gps.config.RedisKeyPrefixConfig;
import com.gzyj.flink.gps.factory.MySQLMybatisSessionFactory;
import com.gzyj.flink.gps.factory.RedisSessionFactory;
import com.gzyj.flink.gps.map.updatetp.UpdateTpDataProcess;
import com.gzyj.flink.gps.mapper.mysql.TpvehdataMapper;
import com.gzyj.flink.gps.sink.gp.StopHistorySink;
import com.gzyj.flink.gps.sink.mysql.TpvehDataSink;
import com.gzyj.flink.pojo.gps.Tpvehdata;
import com.gzyj.flink.pojo.gps.VehStopHistory;
import com.gzyj.flink.utils.PointUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.serialization.SimpleStringSchema;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.util.Collector;
import org.apache.ibatis.session.SqlSession;
import org.apache.ibatis.session.SqlSessionFactory;
import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.data.redis.serializer.StringRedisSerializer;

import java.text.SimpleDateFormat;
import java.util.*;

public class UpdateTpDataFlow {

    private SingleOutputStreamOperator<GPSPosition> gpsSingleStream;

    public UpdateTpDataFlow(SingleOutputStreamOperator<GPSPosition> gpsSingleStream) {
        this.gpsSingleStream = gpsSingleStream;
    }

    public void process() {
        SingleOutputStreamOperator<Tpvehdata> updateSource = this.gpsSingleStream.map(new UpdateTpDataProcess())
                .filter(Objects::nonNull);

        updateSource.filter(new FilterFunction<Tpvehdata>() {
            @Override
            public boolean filter(Tpvehdata tpvehdata) throws Exception {
                return StringUtils.isNotEmpty(tpvehdata.getStoptime()) && tpvehdata.getAcc() == 1;
            }
        }).map(new MapFunction<Tpvehdata, VehStopHistory>() {
            @Override
            public VehStopHistory map(Tpvehdata tpvehdata) throws Exception {


                VehStopHistory history = new VehStopHistory();
                if (tpvehdata.getFlameout() == 0) {
                    history.setVehId(tpvehdata.getVehid());
                    history.setPaiZhao(tpvehdata.getVenumber());
                    history.setYanSe(tpvehdata.getVecolor());
                    history.setDeptId(Integer.valueOf(tpvehdata.getDeptid()));
                    history.setDriverName(tpvehdata.getDriverName());
                    history.setLng(tpvehdata.getLongitude());
                    history.setLat(tpvehdata.getLatitude());
                    history.setStopTime(tpvehdata.getStoptime());
                    history.setEndStopTime(tpvehdata.getTime());
                } else {
                    history.setVehId(tpvehdata.getVehid());
                    history.setPaiZhao(tpvehdata.getVenumber());
                    history.setYanSe(tpvehdata.getVecolor());
                    history.setDeptId(Integer.valueOf(tpvehdata.getDeptid()));
                    history.setDriverName(tpvehdata.getDriverName());
                    history.setStopTime(tpvehdata.getStoptime());
                    history.setFlameOut(tpvehdata.getFlameout());
                    history.setEndStopTime(tpvehdata.getTime());
                }

                return history;
            }
        }).filter(e -> StringUtils.isNotEmpty(e.getVehId())).addSink(new StopHistorySink());


        updateSource.addSink(new TpvehDataSink());


        Properties properties = new Properties();
        properties.setProperty("bootstrap.servers", "127.0.0.1:9092");

        // 算里程
        Time windowSize = Time.minutes(1);

        SingleOutputStreamOperator<GPSPosition> gpsWater = this.gpsSingleStream.assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<GPSPosition>(windowSize) {
            @Override
            public long extractTimestamp(GPSPosition gpsPosition) {
                return gpsPosition.getGPSTimeExt().getTime();
            }
        });
        gpsWater.keyBy(new KeySelector<GPSPosition, String>() {
                    @Override
                    public String getKey(GPSPosition gps) throws Exception {
                        return gps.getVehId();
                    }
                }).window(TumblingProcessingTimeWindows.of(windowSize)).process(new ProcessWindowFunction<GPSPosition, Tpvehdata, String, TimeWindow>() {


                    @Override
                    public void process(String s, ProcessWindowFunction<GPSPosition, Tpvehdata, String, TimeWindow>.Context context, Iterable<GPSPosition> iterable, Collector<Tpvehdata> collector) throws Exception {

                        List<GPSPosition> list = new ArrayList<>();
                        iterable.forEach(e -> list.add(e));
                        Comparator<GPSPosition> nameComparator = Comparator.comparing(GPSPosition::getGPSTimeExt);
                        list.sort(nameComparator);

                        double totalDistance = 0.0;

                        for (int i = 0; i < list.size() - 1; i++) {
                            GPSPosition pre = list.get(i);
                            GPSPosition cur = list.get(i + 1);
                            totalDistance += PointUtils.getDistance(cur.getLatExt(), cur.getLngExt(), pre.getLatExt(), pre.getLngExt());

                        }
                        Long start = context.window().getStart();
                        Long end = context.window().getEnd();
                        Long currentWatermark = context.currentWatermark();
                        SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");
                        Tpvehdata data = new Tpvehdata();
                        data.setVehid(list.get(0).getVehId());
                        data.setDeviceid(list.get(0).getTerminalPhoneNo());
                        data.setTime(list.get(list.size() - 1).getGPSTimeExt());
                        data.setMileToday(totalDistance);
                        collector.collect(data);
                    }
                }).map(new RichMapFunction<Tpvehdata, Tpvehdata>() {
                    SqlSessionFactory mysql;

                    LettuceConnectionFactory redisFactory;

                    private RedisTemplate<String, Object> getRedisTemplate(LettuceConnectionFactory redisSessionFactory) {
                        RedisTemplate<String, Object> redisTemplate = new RedisTemplate<>();
                        redisTemplate.setConnectionFactory(redisSessionFactory);
                        redisTemplate.setKeySerializer(new StringRedisSerializer());
                        redisTemplate.setValueSerializer(new FastJsonRedisSerializer<>(Object.class));
                        redisTemplate.setHashKeySerializer(new StringRedisSerializer());
                        redisTemplate.setHashValueSerializer(new FastJsonRedisSerializer<>(Object.class));
                        redisTemplate.afterPropertiesSet();
                        return redisTemplate;
                    }

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        mysql = MySQLMybatisSessionFactory.getSqlSessionFactory();
                        redisFactory = RedisSessionFactory.getRedisFactory();
                    }

                    public Tpvehdata TryGetData_Last(String simNo, RedisTemplate<String, Object> clusterRedisTemplate) {
                        Object obj = clusterRedisTemplate.opsForValue().get(RedisKeyPrefixConfig.RedisKey_TpData + simNo);
                        if (obj == null) {
                            return null;
                        } else {
                            return JSON.parseObject(JSON.toJSONString(obj, SerializerFeature.DisableCircularReferenceDetect), Tpvehdata.class);
                        }

                    }

                    @Override
                    public Tpvehdata map(Tpvehdata value) throws Exception {
                        RedisTemplate<String, Object> clusterRedisTemplate = getRedisTemplate(redisFactory);
                        SqlSession sqlSession = mysql.openSession(true);
                        Tpvehdata db = TryGetData_Last(value.getDeviceid(), clusterRedisTemplate);

                        try {
                            TpvehdataMapper tpvehdataMapper = sqlSession.getMapper(TpvehdataMapper.class);
                            if(db==null){
                                db = tpvehdataMapper.selectByPrimaryKey(value.getDeviceid());
                            }


                            if (value.getTime().getTime() >= db.getTime().getTime()) {
                                SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd");

                                long startTime = db.getTime().getTime();
                                long endTime = value.getTime().getTime();
                                long diffTime = Math.abs(endTime - startTime);
                                long diffDays = diffTime / (24 * 60 * 60 * 1000);
                                if (diffDays == 0) {
                                    if (db.getMileToday() != null) {
                                        db.setMileToday(db.getMileToday() + value.getMileToday());
                                    } else {
                                        db.setMileToday(value.getMileToday());
                                    }

                                } else {
                                    //跨天
                                    db.setMileToday(0.0);
                                }
                                if (db.getMileTotal() != null) {
                                    db.setMileTotal(db.getMileTotal() + value.getMileToday());
                                } else {
                                    db.setMileTotal(value.getMileToday());
                                }
                                clusterRedisTemplate.opsForValue().set(RedisKeyPrefixConfig.RedisKey_TpData+db.getDeviceid(),db);
                                tpvehdataMapper.insertSelective(db);

                            }


                        } catch (Exception e) {
                            e.printStackTrace();
                            sqlSession.rollback();
                        } finally {
                            sqlSession.close();
                            return db;
                        }


                    }

                }).filter(Objects::nonNull)
                .map(new MapFunction<Tpvehdata, String>() {
                    @Override
                    public String map(Tpvehdata tpvehdata) throws Exception {
                        return JSON.toJSONString(tpvehdata, SerializerFeature.DisableCircularReferenceDetect);
                    }
                }).addSink(new FlinkKafkaProducer<>("gzyj-realtime-gps", new SimpleStringSchema(), properties));

    }

    public static void main(String[] args) {
        SimpleDateFormat dateFormat = new SimpleDateFormat("yyyy-MM-dd");
        try {
            Date startDate = dateFormat.parse("2020-12-31");
            Date endDate = dateFormat.parse("2020-12-30");
            long startTime = startDate.getTime();
            long endTime = endDate.getTime();
            long diffTime = Math.abs(endTime - startTime);
            long diffDays = diffTime / (24 * 60 * 60 * 1000);
            System.out.println("相差天数为：" + diffDays);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }


}
